// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/bitget/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bitget/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"
#include "coin2/exchange/bitget_futures/symbology/product.h"

namespace coin2::exchange::bitget::feed_v1::impl {

using coin2::exchange::bitget_futures::symbology::BitgetFuturesProduct;

std::string ConvertRestSymbol(const std::string& symbol, int64_t timestamp) {
  std::string native_symbol;
  std::vector<std::string> splited = Split(symbol, "_");
  if (splited.size() == 2) {
    native_symbol = splited[0];
  } else if (splited.size() == 3) {
    // quote is USD
    std::string base = splited[0].substr(0, splited[0].size() - 3);
    std::string absolute_norm = base + "-USD.20" + splited[2];
    auto product = BitgetFuturesProduct::FromNormStr(absolute_norm, timestamp);
    native_symbol = product.feed_native_symbol();
  }
  return native_symbol;
}

struct BitgetArguments {
  std::string channel;
  std::string symbol;
  std::string period;
  static BitgetArguments Parse(rapidjson::Document& doc, int64_t timestamp) {
    BitgetArguments args;
    if (doc.HasMember("arg")) {
      args.channel = doc["arg"]["channel"].GetString();
      if (doc["arg"]["channel"].GetString() == std::string_view("books")){
        args.channel = "books_realtime";
      }
      args.symbol = doc["arg"]["instId"].GetString();
    } else if (doc.HasMember("url")) {
      if (strcmp(doc["data"]["code"].GetString(),"00000") != 0) {
        return args;
      }
      std::string url = doc["url"].GetString();
      if (url.find("/v1/kline/getMoreKlineData") != std::string::npos) {
        args.channel = "kline_rest";
        args.symbol = ConvertRestSymbol(doc["payload"]["symbolId"].GetString(), timestamp);
        args.period = doc["payload"]["kLineStep"].GetString();
      } else if (url.find("/api/mix/v1/market/index") != std::string::npos) {
        args.channel = "index";
        auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
        args.symbol = ConvertRestSymbol(parameters["symbol"], timestamp);
      } else if (url.find("/api/mix/v1/market/current-fundRate") != std::string::npos) {
        args.channel = "funding_rate";
        auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
        args.symbol = ConvertRestSymbol(parameters["symbol"], timestamp);
      } else if (url.find("/api/mix/v1/market/funding-time") != std::string::npos) {
        args.channel = "funding_time";
        auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
        args.symbol = ConvertRestSymbol(parameters["symbol"], timestamp);
      } else if (url.find("/api/mix/v1/market/open-interest") != std::string::npos) {
        args.channel = "open_interest";
        auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
        args.symbol = ConvertRestSymbol(parameters["symbol"], timestamp);
      } else if (url.find("/api/mix/v1/market/mark-price") != std::string::npos) {
        args.channel = "mark_price";
        auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
        args.symbol = ConvertRestSymbol(parameters["symbol"], timestamp);
      }
    }
    return args;
  }
};

void BitgetFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  if (packet.payload == "pong") return;

  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("event")) {
    LOG(INFO) << "[Bitget Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  auto args = BitgetArguments::Parse(document, packet.timestamp);

  if (args.symbol.empty()) {
    LOG(INFO) << "[Bitget Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  std::string channel_sign;
  if (args.channel == "kline_rest") {
    channel_sign = "\"kLineStep\":\"" + args.period + "\"";
  } else {
    channel_sign = args.channel;
  }
  if (!CheckChannelFilter(channel_sign)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (!upd_mgr_.is_symbol_registered(args.symbol)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (args.channel == "books5" || args.channel == "books15") {
    auto* book = upd_mgr_.GetBookMutable(args.symbol);
    if (BitgetFeedUpdateParser::ParseBook(book, document, BookBuilder::SNAPSHOT)) {
      upd_mgr_.PublishBook();
    }
  } else if (args.channel == "books_realtime") {
    auto* book = upd_mgr_.GetBookMutable(args.symbol);
    if (parser_config_.publish_bbo_book_only()) {
      std::optional<PriceQty> bid_pre = book->Bid0();
      std::optional<PriceQty> ask_pre = book->Ask0();
      bool book_generated = BitgetFeedUpdateParser::ParseBook(
          book, document, BookBuilder::DELTA);
      std::optional<PriceQty> bid_post = book->Bid0();
      std::optional<PriceQty> ask_post = book->Ask0();
      bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
      if (book_generated && is_bbo_changed) {
        upd_mgr_.PublishBook();
      }
    } else {
      if (BitgetFeedUpdateParser::ParseBook(book, document, BookBuilder::DELTA)) {
        upd_mgr_.PublishBook();
      }
    }
  } else if (args.channel == "trade") {
    int last =  document["data"].GetArray().Size();
    int index = 1;
    for (const auto& trade_json : document["data"].GetArray()) {
      auto* trade = upd_mgr_.GetTradeMutable();
      BitgetFeedUpdateParser::ParseTrade(trade_json, trade);
      upd_mgr_.PublishTrade(args.symbol, index != last);
      index++;
    }
  } else if (args.channel == "kline_rest") {
    auto interval = BitgetFeedUpdateParser::ParseKlineInterval(args.period);
    auto* klines = upd_mgr_.GetKlineMutable(args.symbol, interval);
    if (klines != nullptr) {
      auto data = document["data"]["data"].GetArray();
      std::optional<int64_t> first_open_timestamp;
      if (data.Size() > 0) {
        first_open_timestamp = stoll(data[0][0].GetString()) * 1'000'000LL;
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      for (const auto& kline_data : data) {
        bool freezed = BitgetFeedUpdateParser::ParseKlineRest(kline_data,
            klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      klines->Freeze(); // rest api return freezed data
      if (to_publish) upd_mgr_.PublishKline(args.symbol);
    }
  } else if (args.channel == "index") {
    const auto& msg = document["data"]["data"];
    auto* index = upd_mgr_.GetIndexMutable();
    BitgetFeedUpdateParser::ParseIndex(msg, index);
    upd_mgr_.PublishIndexWithExchangeTimeCheck(args.symbol);
  } else if (args.channel == "funding_rate") {
    const auto& msg = document["data"];
    auto* funding_rate_update = GetFundingRateMutable(args.symbol);
    BitgetFeedUpdateParser::ParseFundingRate(msg, &(funding_rate_update->funding_rate));
    funding_rate_update->funding_rate_update = true;
    funding_rate_update->funding_rate.timestamp = packet.timestamp;
    TryPublishFundingRate(args.symbol, funding_rate_update);
  } else if (args.channel == "funding_time") {
    const auto& msg = document["data"];
    auto* funding_rate_update = GetFundingRateMutable(args.symbol);
    BitgetFeedUpdateParser::ParseFundingTime(msg, &(funding_rate_update->funding_rate));
    funding_rate_update->funding_time_update = true;
    funding_rate_update->funding_rate.timestamp = packet.timestamp;
    TryPublishFundingRate(args.symbol, funding_rate_update);
  } else if (args.channel == "open_interest") {
    auto* open_interest = upd_mgr_.GetOpenInterestMutable();
    BitgetFeedUpdateParser::ParseOpenInterest(document["data"]["data"], open_interest);
    upd_mgr_.PublishOpenInterestWithExchangeTimeCheck(args.symbol);
  } else if (args.channel == "mark_price") {
    auto* mark_price = upd_mgr_.GetMarkPriceMutable();
    BitgetFeedUpdateParser::ParseMarkPrice(document["data"]["data"], mark_price);
    upd_mgr_.PublishMarkPriceWithExchangeTimeCheck(args.symbol);
  } else {
    THROW() << "[Bitget Feed Processor] unkown topic: " << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}

BitgetFeedProcessor::FundingRateUpdate*
BitgetFeedProcessor::GetFundingRateMutable(const std::string& native_symbol) {
  return &symbol2fundingrate_[native_symbol];
}

bool BitgetFeedProcessor::TryPublishFundingRate(
    const std::string& native_symbol,
    FundingRateUpdate* update) {
  if (update->funding_time_update && update->funding_rate_update) {
    auto* funding_rate = upd_mgr_.GetFundingRateMutable();
    *funding_rate = update->funding_rate;
    upd_mgr_.PublishFundingRateWithExchangeTimeCheck(native_symbol);
    update->Clear();
    return true;
  }

  return false;
}

}  // namespace coin2::exchange::bybit_futures::feed_v1::impl
